package cn.tedu.csmall.cart.webapi.kafka;

import cn.tedu.csmall.commons.pojo.cart.model.Cart;
import com.google.gson.Gson;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

// 需要接收kafka的消息,因为kafkaTemple是Spring容器管理的对象
// 所以消息的接收功能,也要保存到Spring容器中
@Component
public class Consumer {

    // SpringKafka接收消息依靠框架提供的"监听机制"
    // 框架中有一个线程,一直实时关注Kafka的消息接收
    // 如果我们指定的话题名称(myCart)接收了消息,那么这条线程就会自动调用下面的方法
    @KafkaListener(topics = "myCart")
    // 下面方法的参数也是来自监听机制,也就是myCart话题接收到的消息
    public void received(ConsumerRecord<String,String> record){
        // 接收消息的方法参数必须是ConsumerRecord
        // 泛型类型和发送消息时指定的泛型一致<[话题名称的类型],[消息的类型]>
        // record就是消息本身,可以从这个对象中获得具体消息内容
        String json=record.value();
        // 将json格式的字符串转换为java对象
        Gson gson=new Gson();
        // 执行转换
        Cart cart=gson.fromJson(json,Cart.class);
        //{"id":2,"commodityCode":"PC100","price":61,"count":9,"userId":"UU100"}
        System.out.println(cart);

    }




}
